package cn.drh.charging.order.mqtt;


import cn.drh.charging.common.pojo.JsonResult;
import cn.drh.charging.common.pojo.param.ChargingProcessParam;
import cn.drh.charging.common.pojo.vo.ChargingProcessVO;
import cn.drh.charging.common.utils.JsonUtils;
import cn.drh.charging.order.constant.Constant;
import cn.drh.charging.order.constant.MqttConstant;
import cn.drh.charging.order.constant.OrderStatusConstant;
import cn.drh.charging.order.pojo.dto.ChargingProcessDto;
import cn.drh.charging.order.pojo.param.ChargingResultParam;
import cn.drh.charging.order.pojo.po.ChargingBillFailPO;
import cn.drh.charging.order.pojo.po.ChargingBillSuccessPO;
import cn.drh.charging.order.pojo.po.ChargingProcessPO;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.BeanUtils;

import java.math.BigDecimal;

/**
 * @ClassName MqttConsumer
 * @Description TODO
 * @Author renhong~Deng
 * @Date 2025/5/28 16:36
 * @Version 1.0
 */

@Slf4j
public class MqttConsumer implements MqttCallbackExtended {
    MqttClient mqttClient;
    MqttContext mqttContext;

    public MqttConsumer(MqttContext mqttContext, MqttClient mqttClient) {
        this.mqttContext = mqttContext;
        this.mqttClient = mqttClient;
    }


    /**
     * 连接成功后触发
     *
     * @param b
     * @param s
     */
    @Override
    public void connectComplete(boolean b, String s) {
        log.debug("MqttConsumer:连接成功");
        try {
            //和EMQX连接成功后,订阅 "/topic/start/result" 来消费消息
            log.debug("连接成功,订阅topic:{},用来接受设备发送的充电结果信息", MqttConstant.TOPIC_CHARGING_RESULT);
            mqttClient.subscribe(MqttConstant.TOPIC_CHARGING_RESULT);
            log.debug("连接成功,订阅topic:{},成功，用来接受设备发送的充电结果信息", MqttConstant.TOPIC_CHARGING_RESULT);

            log.debug("连接成功,订阅topic:{}，用来接收设备同步的充电状态信息", MqttConstant.TOPIC_CHARGING_PROCESS);
            mqttClient.subscribe(MqttConstant.TOPIC_CHARGING_PROCESS);
            log.debug("连接成功,订阅topic:{},用来接收设备同步的充电状态信息", MqttConstant.TOPIC_CHARGING_PROCESS);

        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 连接断开后除非
     *
     * @param cause 失败原因
     */
    @Override
    public void connectionLost(Throwable cause) {
        log.debug("MqttConsumer:连接失败", cause);
    }

    /**
     * 又设备发送消息过来，后触发
     *
     * @param topic   订阅主题
     * @param message 消息内容
     * @throws Exception 转换抛出异常
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.debug("MqttConsumer:消息到达:topic-{},message-{}", topic, message);
        if (MqttConstant.TOPIC_CHARGING_RESULT.equals(topic)) {
            log.debug("处理开始充电结果消息");
            handleChargingResult(message);
        } else if (MqttConstant.TOPIC_CHARGING_PROCESS.equals(topic)) {
            log.debug("处理开始充电进度消息");
            handleChargingProcess(message);
        }
    }

    /**
     * 处理开始充电进度消息
     *
     * @param message
     */
    public void handleChargingProcess(MqttMessage message) {
        log.debug("处理充电进度:{}", message);
        ChargingProcessParam chargingProcessParam = null;
        try {
            chargingProcessParam = JsonUtils.fromJson(message.toString(), ChargingProcessParam.class);
            log.debug("消息转换为java对象:{}", chargingProcessParam);
        } catch (Exception e) {
            log.error("设备发送的充电进度消息格式{}有问题，请检查", message);
        }
        //调用计价服务
        log.debug("通过open-feign的costClient 调用计价服务，计算价格，入参为{}", chargingProcessParam);
        ChargingProcessVO orderCost = getOrderCost(chargingProcessParam);
        log.debug("通过open-feign的costClient 调用计价服务，计算价格，入参为{},出参为{}", chargingProcessParam, orderCost);
        if (orderCost != null) {
            //2 充电进度保存
            saveChargingProcess(chargingProcessParam);
            //3 推送充电数据给用户端 todo
            syncChargingProcess2User(orderCost);

            //4 充电过程检查用户余额
            //钱够
            //钱不够
            //4.1 给用户发通知，钱不够了，引导用户充值，推荐
            //4.2 钱不够 继续充，直到充满为止，记录下用户欠款，等用户下次再充的时候，需要结清欠款
            BigDecimal totalCost = orderCost.getTotalCost();
            log.debug("当前花费金额：{}", totalCost);
            checkUserBalance(chargingProcessParam.getUserId(), totalCost);
            //5.判断温度
            //5.1 大于40度 小于 50 度，B级告警，推送给用户 ，用户可以选择是否继续充
            //5.2 大于50 A级告警，推送给用户，用户可以选择是否继续充
            checkDeviceStatus(chargingProcessParam.getTemperature());
            //6.判断是否充满
            //6.1 设备停止充电  更新订单状态 为订单结束
            checkChargingStatus(chargingProcessParam.getIsFull());
        } else {
            log.error("获取价格失败：{},打印错误日志，告警", chargingProcessParam);
        }
    }

    /**
     * 检查是否充满
     *
     * @param isFull
     */
    private void checkChargingStatus(Boolean isFull) {

    }

    /**
     * 检查设备温度
     *
     * @param temperature 温度
     */
    private void checkDeviceStatus(float temperature) {

    }

    /**
     * 检查用户余额
     *
     * @param userId
     * @param totalCost
     */
    private void checkUserBalance(Integer userId, BigDecimal totalCost) {

    }

    /**
     * 同步充电进度给用户端
     *
     * @param orderCost
     */
    private void syncChargingProcess2User(ChargingProcessVO orderCost) {
        log.debug("同步充电数据到用户手机端：{}", orderCost);
        String json = JsonUtils.toJson(orderCost);
        mqttContext.getWebSocketServer().sendMessage(orderCost.getUserId(), json);

    }

    /**
     * 保存充电进度
     * 数据量很大
     * 设备定时同步 30秒同步一次，一分钟同步2次
     * 1天有1000个用户 充电1000个订单
     * 1个订单充平均4个小时电  有慢充，有快充
     * 1天会产生多少数据？
     * 1天48w数据
     * <p>
     * 1 都把数据存储到表里
     * 2 冷热数据分离  冷就是不经常使用的数据  热就是经常使用的数据
     *
     * @param chargingProcessParam
     */
    private void saveChargingProcess(ChargingProcessParam chargingProcessParam) {
        log.debug("保存充电进度数据:{}", chargingProcessParam);
        ChargingProcessPO chargingProcessPO = new ChargingProcessPO();
        //时间
        chargingProcessPO.setId(System.currentTimeMillis());
        //数据
        BeanUtils.copyProperties(chargingProcessParam, chargingProcessPO);
        //保存到es
        mqttContext.getChargingProcessRepository().save(chargingProcessPO);
        log.debug("保存进度数据成功:{}", chargingProcessParam);
    }

    private ChargingProcessVO getOrderCost(ChargingProcessParam chargingProcessParam) {
        JsonResult<ChargingProcessVO> chargingProcessVOJsonResult =
                mqttContext.getCostClient().calculateCost(chargingProcessParam);
        log.debug("调用计价服务返回结果:{}", chargingProcessVOJsonResult);
        if (chargingProcessVOJsonResult != null) {
            ChargingProcessVO data = chargingProcessVOJsonResult.getData();
            if (data != null) {
                return data;
            }
        }
        return null;
    }

    /**
     * 处理充电结果
     *
     * @param message 消息内容
     */
    public void handleChargingResult(MqttMessage message) {
        //将message转换为对象
        ChargingResultParam deviceResultParam = null;
        try {
            deviceResultParam = JsonUtils.fromJson(message.toString(), ChargingResultParam.class);
            //如果消息不为空，则处理消息
            if (deviceResultParam != null) {
                //取出充电结果 1 开始成功 2 开始失败
                String result = deviceResultParam.getResult();
                if (Constant.RESULT_START_CHARGING_SUCCESS.equals(result)) {
                    //开始充电成功
                    log.debug("充电桩开始充电{}", deviceResultParam);
                    ChargingBillSuccessPO chargingBillSuccessPO = new ChargingBillSuccessPO();
                    chargingBillSuccessPO.setBillId(deviceResultParam.getOrderNo());
                    chargingBillSuccessPO.setBillStatus(OrderStatusConstant.ORDER_STATUS_PROGRESS);
                    log.debug("保存成功订单记录:{}", chargingBillSuccessPO);

                    //如果设备重复发送消息
                    //会导致 成功订单表 charging_bill_success 会重复的保存订单数据
                    //如何解决
                    //1 给订单表 订单编号 创建一个唯一索引,唯一索引的意思 这个表里不能有重复的订单编号
                    // 加完唯一索引后,如果有重复的记录要保存 数据库会抛出异常
                    // Duplicate entry '10000_58_1721720859912' for key 'charging_bill_success_bill_id_unique_key'
                    // 从而避免订单成功表有重复的订单记录
                    // 异常可以处理,可以不处理 通常接入监控系统  监控系统 如果固定的时间之内 重复发送消息,设备出问题了,通知调用方 设备维修人员
                    //2 先查询订单成功表是否有数据,如果没有 插入,如果有 跳过不保存, 代码怎么写? 原子性的问题 todo
                    //3 记录状态 如果处理过了,不再重复处理  redis的布隆过滤器 自己了解 todo
                    try {
                        mqttContext.getOrderRepository().saveSuccessOrder(chargingBillSuccessPO);
                    } catch (Exception e) {
                        log.error("保存成功订单数据失败", e);
                    }
                    log.debug("保存成功订单数据:{},成功", chargingBillSuccessPO);
                } else {
                    //不是成功全是失败
                    log.debug("充电桩开始充电失败,充电桩故障");
                    //创建订单失败记录
                    ChargingBillFailPO chargingBillFailPO = new ChargingBillFailPO();
                    chargingBillFailPO.setBillId(deviceResultParam.getOrderNo());
                    //后续处理无响应的设备 todo
                    chargingBillFailPO.setFailDesc("设备自检失败-设备有响应但是充不了电");
                    log.debug("保存失败订单数据:{}", chargingBillFailPO);
                    //todo 保存失败订单数据 也需要考虑幂等
                    mqttContext.getOrderRepository().saveFailOrder(chargingBillFailPO);
                    log.debug("保存失败订单数据:{},成功", chargingBillFailPO);
                }
            }
        } catch (Exception e) {
            log.error("设备返回消息内容有问题", e);
        }
    }


    /**
     * 消息发送完成触发
     *
     * @param iMqttDeliveryToken
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        log.debug("MqttConsumer:消息处理完成");
    }
}
